package org.workstream.processnode;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicLong;

import org.workstream.data.DataNode;
import org.workstream.driver.LocalDriver;


public abstract class DataProcessListNode {
	protected Queue<DataNode> zQueue;
	private final int pthread_num;
	private Thread[] zThreads;
	private boolean suspend;
	private Object suspend_mutex;
	private boolean threadStopFlag;
	private AtomicLong processCapacity;
	protected HashSet<Class<? extends DataNode>> produceDataTypeSet;
	protected HashSet<Class<? extends DataNode>> dealDataTypeSet;
	private LocalDriver driver = null;
	private static int refCount;
	static{
		refCount = 1;
	}
	public boolean isSuspend() {
		return suspend;
	}
	public Object getSuspend_mutex() {
		return suspend_mutex;
	}
	public int getQueueSize(){
		return zQueue.size();
	}
	public boolean getThreadStopFlag(){
		return threadStopFlag;
	}
	
	public LocalDriver getDriver() {
		return driver;
	}
	public void setDriver(LocalDriver driver) {
		this.driver = driver;
	}
	public DataProcessListNode(final int pthread_num,LocalDriver driver,HashSet<Class<? extends DataNode>> produceDataTypeSet,HashSet<Class<? extends DataNode>> dealDataTypeSet){
		if(pthread_num <= 0)
			throw new IllegalArgumentException("ZDataProcessListNode constructor wrong arguments");
		processCapacity = new AtomicLong(0);
		suspend = false;
		suspend_mutex = new Object();
		zQueue = new LinkedList<DataNode>();
		threadStopFlag = false;
		this.pthread_num = pthread_num;
		this.driver = driver;
		this.produceDataTypeSet = produceDataTypeSet;
		this.dealDataTypeSet = dealDataTypeSet;
		zThreads = new Thread[pthread_num];
		for(int i = 0; i < pthread_num; ++i){
			zThreads[i] = new Thread(){
				@Override
				public void run() {
					DataNode node = null;
					while(true){
						node = null;
						synchronized(suspend_mutex){
							while(suspend){
								//System.out.println("suspend");
								try {
									suspend_mutex.wait(5000);
								} catch (InterruptedException e) {
									// TODO Auto-generated catch block
									e.printStackTrace();
								}finally{
									if(threadStopFlag)
										break;
								}
							}
						}
						synchronized(zQueue){
							while(zQueue.size() <= 0){
								try {
									zQueue.wait(5000);
								} catch (InterruptedException e) {
								}
								finally{
									if(zQueue.size() <= 0 && threadStopFlag)
										break;
								}
							}
							if(zQueue.size() <= 0 && threadStopFlag)
								break;
							node = zQueue.poll();
						}
						if(node != null){
							DataProcessListNode.this.driver.DescAndGetTotalQueueSize(node);
							incrAndGetPCapacity();
							processData(node);
						}
					}
				}
			};
			zThreads[i].start();
			this.driver.RegistPNodeThread(zThreads[i], this);
		}
	}
	public long getPCapacity(){
		return processCapacity.get();
	}
	public long getAndIncrPCapacity(){
		return processCapacity.getAndIncrement();
	}
	public long incrAndGetPCapacity(){
		return processCapacity.incrementAndGet();
	}
	public long getAndDescPCapacity(){
		return processCapacity.getAndDecrement();
	}
	public long descAndGetPCapacity(){
		return processCapacity.decrementAndGet();
	}
	public void addTaskNode(DataNode node){
		synchronized(zQueue){
			zQueue.offer(node);
			zQueue.notifyAll();
		}
	}
	public void suspendAllThreads(){
		synchronized(suspend_mutex){
			if(!suspend){
				if(getPCapacity() < 0){
					//System.out.println(this.getClass().getSimpleName() +": suspendAllThreads");
					suspend = true;
				}
			}
		}
	}
	public void resumeAllThreads(){
		synchronized(suspend_mutex){
			if(suspend){
				//System.out.println(this.getClass().getSimpleName() +": resumeAllThreads");
				suspend = false;
				suspend_mutex.notifyAll();
			}
		}
	}
	protected abstract void processData(DataNode node);
	public void close(){
		threadStopFlag = true;
		for(int i = 0; i < zThreads.length; ++i){
			zThreads[i].interrupt();
			try {
				zThreads[i].join();
			} catch (InterruptedException e) {
			}
		}
	}
	public HashSet<Class<? extends DataNode>> getDealDataTypeSet() {
		return dealDataTypeSet;
	}
	public HashSet<Class<? extends DataNode>> getProduceDataTypeSet() {
		return produceDataTypeSet;
	}
	public void addRefCount(){
		++refCount;
	}
	public int subRefCount(){
		return --refCount;
	}
	public Queue<DataNode> getzQueue() {
		return zQueue;
	}
	
}
